package org.common.utils.rpc.core;
import org.common.utils.hash.ConsistentHash;
import org.common.utils.net.netty.NettyClient;
import org.common.utils.rpc.exception.RpcNotActivityClientException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * RPC连接
 * @author zhouzhibing
 * @date 2025/11/15 23:34
 */
public class RpcConnection {
    private static final Logger log = LoggerFactory.getLogger(RpcConnection.class);
    /**
     * 活跃的连接列表
     */
    private final Map<Long, NettyClient> activityClients = new ConcurrentHashMap<>();
    /**
     * 丢失的连接列表
     */
    private final Map<Long, NettyClient> loseClients = new ConcurrentHashMap<>();
    /**
     * 客户端连接的，一致性哈希对象。
     */
    private ConsistentHash<NettyClient> consistentHash;

    /**
     * 添加连接
     * @param nettyClient 连接客户端
     */
    public void addClient(NettyClient nettyClient) {
        nettyClient.start();
        nettyClient.setEventListener(this::triggerEvent);
        log.info("rpc client {} add" , nettyClient);
    }

    /**
     * 根据对象toString hash获取对象，如果对象为空，则使用当前时间戳作为key
     * @param key 参数对象
     * @return 连接对象
     */
    public NettyClient getClient(String key) {
        if(activityClients.isEmpty()) {
            throw new RpcNotActivityClientException();
        }

        if(activityClients.size() == 1) {
            return activityClients.values().iterator().next();
        }
        return consistentHash.get(key);
    }

    /**
     * 触发客户端事件
     * @param client 连接对象
     * @param event 事件枚举对象
     */
    private void triggerEvent(NettyClient client , NettyClient.Event event) {
        if (event == NettyClient.Event.CONNECT_SUCCESS) {
            activityClient(client);
        } else {
            loseClient(client);
        }
    }

    /**
     * 丢失连接
     * @param client 连接
     */
    private void loseClient(NettyClient client) {
        RpcAttach attach = client.getAttach();
        long connectionId = attach.getId();

        // 如果已经丢失，则不处理
        if (loseClients.containsKey(connectionId)) {
            return ;
        }

        loseClients.put(connectionId , client);
        activityClients.remove(connectionId);

        // 如果连接数大于1，则创建一致性哈希对象
        if(activityClients.size() > 1) {
            consistentHash = newConsistentHash();
        }
        log.info("rpc client {} lose" , client);
    }

    /**
     * 活跃连接
     * @param client 连接
     */
    private void activityClient(NettyClient client) {
        RpcAttach attach = client.getAttach();
        long connectionId = attach.getId();

        // 如果已经活跃，则不处理
        if (activityClients.containsKey(connectionId)) {
            return ;
        }

        loseClients.remove(connectionId);
        activityClients.put(connectionId , client);

        // 如果连接数大于1，则创建一致性哈希对象
        if(activityClients.size() > 1) {
            consistentHash = newConsistentHash();
        }
        log.info("rpc client {} activity" , client);
    }

    /**
     * 创建一致性哈希对象
     * @return 一致性哈希对象
     */
    private ConsistentHash<NettyClient> newConsistentHash() {
        ConsistentHash<NettyClient> consistentHash = new ConsistentHash<>();
        for (NettyClient client : activityClients.values()) {
            consistentHash.add(client);
        }
        return consistentHash;
    }
}
